//
// Created by zy on 22-7-8.
//
//
// Created by zy on 22-6-7.
//
#include "db/version_set.h"
#include <cassert>
#include <chrono>
#include <cmath>
#include <cstring>
#include <fstream>
#include <iostream>
#include <random>
#include <unistd.h>

#include "leveldb/comparator.h"
#include "leveldb/db.h"

#include "mod/cxxopts.hpp"
#include "mod/learned_index.h"
#include "mod/stats.h"
#include "mod/util.h"

using namespace std::chrono;
using namespace leveldb;
using namespace adgMod;
using std::string;
using std::cerr;
using std::cout;
using std::endl;
using std::to_string;
using std::vector;
using std::map;
using std::ifstream;
using std::string;

int num_pairs_base = 1000;
int mix_base = 20;
#include <set>


class NumericalComparator : public Comparator {
 public:
  NumericalComparator() = default;
  virtual const char* Name() const {return "adgMod:NumericalComparator";}
  virtual int Compare(const Slice& a, const Slice& b) const {
    uint64_t ia = adgMod::ExtractInteger(a.data(), a.size());
    uint64_t ib = adgMod::ExtractInteger(b.data(), b.size());
    if (ia < ib) return -1;
    else if (ia == ib) return 0;
    else return 1;
  }
  virtual void FindShortestSeparator(std::string* start, const Slice& limit) const { return; };
  virtual void FindShortSuccessor(std::string* key) const { return; };
};


void PutAndPrefetch(int lower, int higher, vector<string>& keys) {
  adgMod::Stats* instance = adgMod::Stats::GetInstance();

  Status status;

  instance->StartTimer(9);
  for (int i = lower; i < higher; ++i) {
    string value = generate_value(0);
    status = db->Put(write_options, keys[i], value);
    assert(status.ok() && "File Put Error");
  }
  instance->PauseTimer(9, true);

  //cout << "Put Complete" << endl;


  instance->StartTimer(10);
  for (int i = lower; i < higher; ++i) {
    string value;
    status = db->Get(read_options, keys[i], &value);
    //cout << "Get " << keys[i] << " Done" << endl;
    assert(status.ok() && "File Get Error");
  }
  instance->PauseTimer(10, true);

  //cout << "Prefetch Complete" << endl;
};


int main(int argc, char *argv[]) {
  int num_operations, num_iteration, num_mix;
  float test_num_segments_base;
  float num_pair_step;
  string db_location, profiler_out, input_filename, distribution_filename,
      ycsb_filename;
  bool print_single_timing, print_file_info, evict, unlimit_fd,
      use_distribution = false, pause, use_ycsb = false;
  bool change_level_load, change_file_load, change_level_learning,
      change_file_learning;
  int insert_bound, length_range;
  string db_location_copy;

  cxxopts::Options commandline_options("leveldb read test",
                                       "Testing leveldb read performance.");
  commandline_options.add_options()(
      "n,get_number", "the number of gets (to be multiplied by 1024)",
      cxxopts::value<int>(num_operations)->default_value("1000"))(
      "s,step", "the step of the loop of the size of db",
      cxxopts::value<float>(num_pair_step)->default_value("1"))(
      "i,iteration", "the number of iterations of a same size",
      cxxopts::value<int>(num_iteration)->default_value("1"))(
      "m,modification", "if set, run our modified version",
      cxxopts::value<int>(adgMod::MOD)->default_value("0"))(
      "h,help", "print help message",
      cxxopts::value<bool>()->default_value("false"))(
      "d,directory", "the directory of db",
      cxxopts::value<string>(db_location)->default_value("/mnt/ssd/testdb"))(
      "k,key_size", "the size of key",
      cxxopts::value<int>(adgMod::key_size)->default_value("8"))(
      "v,value_size", "the size of value",
      cxxopts::value<int>(adgMod::value_size)->default_value("8"))(
      "single_timing", "print the time of every single get",
      cxxopts::value<bool>(print_single_timing)->default_value("false"))(
      "file_info", "print the file structure info",
      cxxopts::value<bool>(print_file_info)->default_value("false"))(
      "test_num_segments", "test: number of segments per level",
      cxxopts::value<float>(test_num_segments_base)->default_value("1"))(
      "string_mode", "test: use string or int in model",
      cxxopts::value<bool>(adgMod::string_mode)->default_value("false"))(
      "e,model_error", "error in modesl",
      cxxopts::value<uint32_t>(adgMod::model_error)->default_value("8"))(
      "f,input_file", "the filename of input file",
      cxxopts::value<string>(input_filename)->default_value(""))(
      "multiple", "test: use larger keys",
      cxxopts::value<uint64_t>(adgMod::key_multiple)->default_value("1"))(
      "w,write", "writedb",
      cxxopts::value<bool>(fresh_write)->default_value("false"))(
      "c,uncache", "evict cache",
      cxxopts::value<bool>(evict)->default_value("false"))(
      "u,unlimit_fd", "unlimit fd",
      cxxopts::value<bool>(unlimit_fd)->default_value("false"))("x,dummy",
                                                                "dummy option")(
      "filter", "use filter",
      cxxopts::value<bool>(adgMod::use_filter)->default_value("false"))(
      "mix", "mix read and write",
      cxxopts::value<int>(num_mix)->default_value("0"))(
      "distribution", "operation distribution",
      cxxopts::value<string>(distribution_filename)->default_value(""))(
      "change_level_load", "load level model",
      cxxopts::value<bool>(change_level_load)->default_value("false"))(
      "change_file_load", "enable level learning",
      cxxopts::value<bool>(change_file_load)->default_value("false"))(
      "change_level_learning", "load file model",
      cxxopts::value<bool>(change_level_learning)->default_value("false"))(
      "change_file_learning", "enable file learning",
      cxxopts::value<bool>(change_file_learning)->default_value("false"))(
      "p,pause", "pause between operation",
      cxxopts::value<bool>(pause)->default_value("false"))(
      "policy", "learn policy",
      cxxopts::value<int>(adgMod::policy)->default_value("0"))(
      "YCSB", "use YCSB trace",
      cxxopts::value<string>(ycsb_filename)->default_value(""))(
      "insert", "insert new value",
      cxxopts::value<int>(insert_bound)->default_value("0"))(
      "range", "use range query and specify length",
      cxxopts::value<int>(length_range)->default_value("0"));
  auto result = commandline_options.parse(argc, argv);
  if (result.count("help")) {
    printf("%s", commandline_options.help().c_str());
    exit(0);
  }

  std::default_random_engine e1(0), e2(255);
  srand(0);
  num_operations *= num_pairs_base;  // 1000 * 1000,什么意思呢？

  adgMod::fd_limit = unlimit_fd ? 1024 * 1024 : 1024;  // TODO: 需要测试一下是否真的影响性能
  adgMod::restart_read = true;
  adgMod::level_learning_enabled ^= change_level_learning;
  adgMod::file_learning_enabled ^= change_file_learning;
  adgMod::load_level_model ^= change_level_load;
  adgMod::load_file_model ^= change_file_load;

  // adgMod::file_learning_enabled = false;

  vector<uint64_t> distribution;  // uint64_t是unsigned long int的typedef别名
  vector<int> ycsb_is_write;
  // keys.reserve(100000000000 / adgMod::value_size);

  //  std::cerr << "读取keys from " << input_filename << std::endl;
  ifstream input(input_filename);
  assert(input.good());  // 加个判定，提前结束错误的情况
  std::string dev_id;
  char* op_flag;
  std::string key;
  vector<string> keys_all;
  vector<char> ops_all;

  string line;
  while (getline(input, line)) {
    vector<string> data_line;
    string number;  // 一行里的每个用逗号分隔的部分
    std::istringstream readstr(line);
    // 将一行数据按逗号分割
    for (int j=0; j<5; j++) {
      getline(readstr, number, ',');
      data_line.push_back(number);
    }
    op_flag = (char*) (data_line[1]).data();
    key = data_line[2];
    string the_key = generate_key(key);
    keys_all.push_back(std::move(the_key));
    ops_all.push_back(std::move(*op_flag));
  }

  int N = int(keys_all.size() * 3.0/5.0);

  std::vector<std::string> keys_former(keys_all.begin(), keys_all.begin() + N);
  std::vector<std::string> keys_latter(keys_all.begin() + N, keys_all.end());
  std::vector<char> ops_former(ops_all.begin(), ops_all.begin() + N);
  std::vector<char> ops_latter(ops_all.begin() + N, ops_all.end());

  keys_all.clear();
  ops_all.clear();
  //  std::cerr << input_filename << "-real-finish keys" << std::endl;

  // 在load_ar的时候copy_out是false
  adgMod::Stats* instance = adgMod::Stats::GetInstance();
  vector<vector<size_t>> times(20);
  string values(1024 * 1024, '0');

  //  std::cerr << "正在执行drop_caches操作：清除页缓存和inode、dentry缓存" << std::endl;
  //  system("sync; echo 3 | sudo tee /proc/sys/vm/drop_caches");
  int iteration = 0;
  std::uniform_int_distribution<uint64_t> uniform_dist_value(0, (uint64_t)values.size() - adgMod::value_size - 1);

  // 这里先把前半个数据集写入
  DB* db;
  Options options;
  ReadOptions& read_options = adgMod::read_options;
  WriteOptions& write_options = adgMod::write_options;
  Status status;
  options.create_if_missing = true;
  write_options.sync = false;
  instance->ResetAll();

  // 写入开始
  string command = "rm -rf " + db_location;
  system(command.c_str());
  //      std::cerr << "执行fstrim操作：生成sstable文件" << std::endl;
//  system("sudo fstrim -a -v");
  //      std::cerr << "开始写入leveldb：生成sstable文件" << std::endl;

  status = DB::Open(options, db_location, &db);
  assert(status.ok() && "Open Error");

  instance->StartTimer(9);
  std::vector<std::string> keys = keys_former;
  assert(keys.size() == ops_former.size());
  int cut_size;
  // 对keys.size分情况讨论还是很有必要的，因为key的数量可能很少，不到10万。
  if (keys.size() > 100000) {
    cut_size = keys.size() / 100000;
  } else {
    cut_size = keys.size() / 1000;
  }
  //      std::cerr << "cut_size: " << cut_size << std::endl;
  std::vector<std::pair<int, int>> chunks;

  for (int cut = 0; cut < cut_size; ++cut) {
    chunks.emplace_back(keys.size() * cut / cut_size,
                        keys.size() * (cut + 1) / cut_size);
  }

  // 加载数据库（放置数据，not ssd kind of），加载完了Bourbon才能学
  for (int cut = 0; cut < chunks.size(); ++cut) {
    //        std::cerr << "leveldb writing process: " << cut << " / " << chunks.size() << "\r";
    for (int i = chunks[cut].first; i < chunks[cut].second; ++i) {
      if (ops_former[i] == 'W') {
        status = db->Put(write_options, keys[i],
                         {values.data() + uniform_dist_value(e2),
                          (uint64_t)adgMod::value_size});
        assert(status.ok() && "File Put Error");
      } else if (ops_former[i] == 'R') {
        std::string value_load;
        db->Get(read_options, keys[i], &value_load);
        //            assert(status.ok() && "File Get Error");  // 这里不断言，因为不一定能找到，而且也不用打印not found
      } else throw "wrong operation flag";
    }
  }
  adgMod::db->vlog->Sync();
  instance->PauseTimer(9, true);
  std::cout << "Put Complete" << std::endl;
  // todo :上面加载的到底是leveldb还是wisckey?
  //      std::cerr << input_filename << "-real-finish leveldb" << std::endl;
  keys.clear();

  if (print_file_info && iteration==0)  db->PrintFileInfo();
  adgMod::db->WaitForBackground();
  sleep(5);
  delete db;

  // 第二阶段，开始建立学习索引。注意建立学习索引是跟加载数据库一一对应的
  status = DB::Open(options, db_location, &db);
  adgMod::db->WaitForBackground();

  if (adgMod::MOD == 6 || adgMod::MOD == 7) {
    Version* current = adgMod::db->versions_->current();
    for (int i = 1; i < config::kNumLevels; ++i) {
      LearnedIndexData::Learn(new VersionAndSelf{current, adgMod::db->version_count,
                                                 current->learned_index_data_[i].get(), i});
    }
    current->FileLearn();
  }

  std::cout << "Shutting down" << std::endl;
  adgMod::db->WaitForBackground();
  delete db;

  std::string data_name_4print = get_dataname_from_filepath(input_filename);
  std::cerr << data_name_4print << "-real-load-done" << std::endl;

}